package com.ksyun.campus.metaserver.services.impl;

import com.ksyun.campus.metaserver.client.DataServerClient;
import com.ksyun.campus.metaserver.config.MinfsProperties;
import com.ksyun.campus.metaserver.domain.BlockInfo;
import com.ksyun.campus.metaserver.domain.DataServerInfo;
import com.ksyun.campus.metaserver.services.IFsckService;
import com.ksyun.campus.metaserver.storage.MetadataStore;
import com.ksyun.campus.metaserver.zk.ZooKeeperService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 文件系统检查和修复服务实现类
 * 负责执行文件系统一致性检查和自动修复操作
 * 包括检测和修复块副本不足、过多副本以及其他文件系统异常
 */
@Slf4j
@Service
public class FsckServiceImpl implements IFsckService {
    
    /**
     * 元数据存储服务，用于获取和更新块信息
     */
    @Autowired
    private MetadataStore metadataStore;
    
    /**
     * ZooKeeper服务，用于获取当前DataServer列表和判断当前节点是否为主节点
     */
    @Autowired
    private ZooKeeperService zooKeeperService;
    
    /**
     * DataServer客户端，用于与DataServer通信执行块复制和删除操作
     */
    @Autowired
    private DataServerClient dataServerClient;
    
    /**
     * 系统配置属性，包含FSCK相关配置
     */
    @Autowired
    private MinfsProperties properties;
    
    /**
     * FSCK历史记录，保存所有执行过的FSCK操作的统计信息
     */
    private final Map<String, IFsckService.FsckStats> fsckHistory = new ConcurrentHashMap<>();
    
    /**
     * 最新的FSCK统计信息
     */
    private volatile IFsckService.FsckStats latestStats;

    /**
     * 定时执行FSCK任务
     * 根据配置的时间间隔自动执行文件系统检查
     * 只有在当前节点为主节点且FSCK功能已启用时才会执行
     */
    @Scheduled(fixedRateString = "${minfs.fsck.interval-seconds:60}000")
    public void fsckTask() {
        if (!zooKeeperService.isLeader()) {
            log.debug("Not leader, skipping FSCK");
            return;
        }
        
        if (!properties.getFsck().isEnabled()) {
            log.debug("FSCK disabled, skipping");
            return;
        }
        
        runFsck();
    }
    
    /**
     * 运行文件系统检查和修复
     * 执行全面的文件系统检查，包括：
     * 1. 检查所有块的副本数量是否符合要求
     * 2. 检查DataServer健康状态
     * 3. 修复副本不足的块（增加副本）
     * 4. 修复副本过多的块（删除多余副本）
     *
     * @return FSCK统计信息，包含检查和修复的详细结果
     */
    @Override
    public IFsckService.FsckStats runFsck() {
        log.info("Starting FSCK scan");
        
        IFsckService.FsckStats stats = new IFsckService.FsckStats();
        stats.setStartTime(LocalDateTime.now());
        
        try {
            // 获取所有块和健康的DataServer
            List<BlockInfo> allBlocks = metadataStore.getAllBlocks();
            List<DataServerInfo> healthyDataServers = zooKeeperService.getCurrentDataServers().stream()
                    .filter(ds -> ds.isHealthy(properties.getHeartbeat().getTimeoutSeconds()))
                    .collect(Collectors.toList());
            
            Set<String> healthyDsIds = healthyDataServers.stream()
                    .map(DataServerInfo::getDsId)
                    .collect(Collectors.toSet());
            
            stats.setTotalBlocks(allBlocks.size());
            stats.setHealthyDataServers(healthyDataServers.size());
            
            log.info("FSCK: Found {} blocks and {} healthy data servers", 
                    allBlocks.size(), healthyDataServers.size());
            
            // 检查每个块
            for (BlockInfo block : allBlocks) {
                checkAndRepairBlock(block, healthyDsIds, healthyDataServers, stats);
            }
            
        } catch (Exception e) {
            log.error("Error during FSCK", e);
            stats.setErrorMessage(e.getMessage());
        }
        
        stats.setEndTime(LocalDateTime.now());
        stats.setDurationMs(java.time.Duration.between(stats.getStartTime(), stats.getEndTime()).toMillis());
        
        // 更新统计信息
        latestStats = stats;
        fsckHistory.put(stats.getStartTime().toString(), stats);
        
        log.info("FSCK completed. Checked: {}, Under-replicated: {}, Over-replicated: {}, Corrupted: {}, Repaired: {}, Failed: {}", 
                stats.getTotalBlocks(), stats.getUnderReplicatedBlocks(), 
                stats.getOverReplicatedBlocks(), stats.getCorruptedBlocks(), 
                stats.getRepairedBlocks(), stats.getFailedRepairs());
        
        return stats;
    }
    
    /**
     * 检查并修复单个块
     * 检查块的副本数量是否符合要求，并根据需要进行修复
     * 现在会实际验证块文件是否存在，而不仅仅依据心跳判断
     *
     * @param block 要检查的块信息
     * @param healthyDsIds 健康DataServer的ID集合
     * @param healthyDataServers 健康DataServer列表
     * @param stats FSCK统计信息，用于记录检查和修复结果
     */
    private void checkAndRepairBlock(BlockInfo block, Set<String> healthyDsIds, 
                                   List<DataServerInfo> healthyDataServers, IFsckService.FsckStats stats) {
        try {
            // 筛选健康的副本，并验证块文件实际存在
            List<BlockInfo.ReplicaInfo> actuallyHealthyReplicas = block.getReplicas().stream()
                    .filter(replica -> healthyDsIds.contains(replica.getDsId()))
                    .filter(replica -> verifyBlockExists(replica, block, healthyDataServers))
                    .collect(Collectors.toList());
            
            int targetReplication = properties.getCluster().getReplicationFactor();
            int currentActualReplicas = actuallyHealthyReplicas.size();
            int originalReplicaCount = block.getReplicas().stream()
                    .filter(replica -> healthyDsIds.contains(replica.getDsId()))
                    .collect(Collectors.toList()).size();
            
            // 如果实际存在的副本数少于元数据中记录的副本数，记录差异
            if (currentActualReplicas < originalReplicaCount) {
                log.warn("Block {}:{} has {} replicas in metadata but only {} actually exist", 
                        block.getFileId(), block.getBlkIdx(), originalReplicaCount, currentActualReplicas);
                stats.incrementCorruptedBlocks();
            }
            
            if (currentActualReplicas < targetReplication) {
                // 副本不足
                stats.incrementUnderReplicatedBlocks();
                repairUnderReplicatedBlock(block, actuallyHealthyReplicas, healthyDataServers, targetReplication, stats);
                
            } else if (currentActualReplicas > targetReplication) {
                // 副本过多
                stats.incrementOverReplicatedBlocks();
                removeExcessReplicas(block, actuallyHealthyReplicas, targetReplication, stats);
            }
            
        } catch (Exception e) {
            log.error("Error checking block {}:{}", block.getFileId(), block.getBlkIdx(), e);
            stats.incrementFailedRepairs();
        }
    }
    
    /**
     * 验证副本对应的块文件是否实际存在
     * 
     * @param replica 副本信息
     * @param block 块信息
     * @param healthyDataServers 健康DataServer列表
     * @return true如果块文件存在，false否则
     */
    private boolean verifyBlockExists(BlockInfo.ReplicaInfo replica, BlockInfo block, 
                                    List<DataServerInfo> healthyDataServers) {
        try {
            // 找到对应的DataServer
            DataServerInfo dataServer = healthyDataServers.stream()
                    .filter(ds -> ds.getDsId().equals(replica.getDsId()))
                    .findFirst()
                    .orElse(null);
            
            if (dataServer == null) {
                log.debug("DataServer {} not found for replica verification", replica.getDsId());
                return false;
            }
            
            // 通过DataServerClient检查块是否存在
            boolean exists = dataServerClient.blockExists(dataServer, block.getFileId(), block.getBlkIdx());
            
            if (!exists) {
                log.warn("Block {}:{} does not exist on DataServer {} despite being in metadata", 
                        block.getFileId(), block.getBlkIdx(), replica.getDsId());
            }
            
            return exists;
            
        } catch (Exception e) {
            log.error("Error verifying block existence {}:{} on DataServer {}", 
                    block.getFileId(), block.getBlkIdx(), replica.getDsId(), e);
            return false;
        }
    }
    
    /**
     * 修复副本不足的块
     * 通过从现有健康副本复制数据到新选择的DataServer来增加副本数量
     * 同时从元数据中移除已损坏的副本记录
     *
     * @param block 需要修复的块信息
     * @param healthyReplicas 当前健康的副本列表
     * @param healthyDataServers 可用的健康DataServer列表
     * @param targetReplication 目标副本数量
     * @param stats FSCK统计信息，用于记录修复结果
     */
    private void repairUnderReplicatedBlock(BlockInfo block, List<BlockInfo.ReplicaInfo> healthyReplicas,
                                          List<DataServerInfo> healthyDataServers, int targetReplication, IFsckService.FsckStats stats) {
        if (healthyReplicas.isEmpty()) {
            log.warn("Block {}:{} has no healthy replicas, cannot repair", 
                    block.getFileId(), block.getBlkIdx());
            stats.incrementFailedRepairs();
            return;
        }
        
        // 选择源副本（优先选择主副本）
        BlockInfo.ReplicaInfo sourceReplica = healthyReplicas.stream()
                .filter(r -> r.getRole() == BlockInfo.ReplicaInfo.ReplicaRole.PRIMARY)
                .findFirst()
                .orElse(healthyReplicas.get(0));
        
        DataServerInfo sourceDs = findDataServerById(sourceReplica.getDsId(), healthyDataServers);
        if (sourceDs == null) {
            log.warn("Source DataServer {} not found for block {}:{}", 
                    sourceReplica.getDsId(), block.getFileId(), block.getBlkIdx());
            stats.incrementFailedRepairs();
            return;
        }
        
        // 计算需要添加的副本数量
        int replicasToAdd = targetReplication - healthyReplicas.size();
        Set<String> existingDsIds = healthyReplicas.stream()
                .map(BlockInfo.ReplicaInfo::getDsId)
                .collect(Collectors.toSet());
        
        // 选择目标DataServer
        List<DataServerInfo> targetDataServers = selectTargetDataServers(
                healthyDataServers, existingDsIds, replicasToAdd, block.getFileId(), block.getBlkIdx());
        
        // 创建新的副本列表，先只包含实际健康的副本
        List<BlockInfo.ReplicaInfo> newReplicas = new ArrayList<>(healthyReplicas);
        
        // 复制到目标DataServer
        for (DataServerInfo targetDs : targetDataServers) {
            boolean success = dataServerClient.replicateBlock(
                    sourceDs, targetDs, block.getFileId(), block.getBlkIdx(), block.getSize());
            
            if (success) {
                // 添加新副本到元数据
                BlockInfo.ReplicaInfo newReplica = BlockInfo.ReplicaInfo.builder()
                        .dsId(targetDs.getDsId())
                        .host(targetDs.getHost())
                        .port(targetDs.getPort())
                        .role(BlockInfo.ReplicaInfo.ReplicaRole.REPLICA)
                        .build();
                
                newReplicas.add(newReplica);
                stats.incrementRepairedBlocks();
                
                log.info("Successfully replicated block {}:{} to DataServer {}", 
                        block.getFileId(), block.getBlkIdx(), targetDs.getDsId());
            } else {
                stats.incrementFailedRepairs();
                log.error("Failed to replicate block {}:{} to DataServer {}", 
                        block.getFileId(), block.getBlkIdx(), targetDs.getDsId());
            }
        }
        
        // 更新块元数据，移除损坏的副本并添加新的副本
        if (newReplicas.size() != block.getReplicas().size() || newReplicas.size() < targetReplication) {
            try {
                BlockInfo updatedBlock = BlockInfo.builder()
                        .fileId(block.getFileId())
                        .blkIdx(block.getBlkIdx())
                        .size(block.getSize())
                        .replicas(newReplicas)
                        .build();
                
                metadataStore.putBlock(block.getFileId(), block.getBlkIdx(), updatedBlock);
                
                log.info("Updated metadata for block {}:{} - removed corrupted replicas, now has {} replicas", 
                        block.getFileId(), block.getBlkIdx(), newReplicas.size());
                        
            } catch (Exception e) {
                log.error("Failed to update block metadata for {}:{}", 
                        block.getFileId(), block.getBlkIdx(), e);
            }
        }
    }
    
    /**
     * 移除多余的副本
     * 删除超出目标副本数量的副本，优先保留主副本
     *
     * @param block 需要处理的块信息
     * @param healthyReplicas 当前健康的副本列表
     * @param targetReplication 目标副本数量
     * @param stats FSCK统计信息，用于记录处理结果
     */
    private void removeExcessReplicas(BlockInfo block, List<BlockInfo.ReplicaInfo> healthyReplicas,
                                    int targetReplication, IFsckService.FsckStats stats) {
        // 按优先级排序副本（保留主副本，然后按负载/偏好排序）
        healthyReplicas.sort((r1, r2) -> {
            if (r1.getRole() == BlockInfo.ReplicaInfo.ReplicaRole.PRIMARY) return -1;
            if (r2.getRole() == BlockInfo.ReplicaInfo.ReplicaRole.PRIMARY) return 1;
            return 0; // 可以在这里添加更复杂的排序逻辑
        });
        
        // 移除多余的副本
        List<BlockInfo.ReplicaInfo> toRemove = healthyReplicas.subList(targetReplication, healthyReplicas.size());
        
        for (BlockInfo.ReplicaInfo replica : toRemove) {
            DataServerInfo ds = findDataServerById(replica.getDsId(), 
                    zooKeeperService.getCurrentDataServers());
            
            if (ds != null) {
                boolean success = dataServerClient.deleteBlock(ds, block.getFileId(), block.getBlkIdx());
                if (success) {
                    log.info("Removed excess replica of block {}:{} from DataServer {}", 
                            block.getFileId(), block.getBlkIdx(), ds.getDsId());
                } else {
                    log.warn("Failed to remove excess replica of block {}:{} from DataServer {}", 
                            block.getFileId(), block.getBlkIdx(), ds.getDsId());
                }
            }
        }
        
        // 更新元数据
        try {
            List<BlockInfo.ReplicaInfo> allReplicas = new ArrayList<>(block.getReplicas());
            allReplicas.removeIf(r -> toRemove.stream().anyMatch(tr -> tr.getDsId().equals(r.getDsId())));
            
            BlockInfo updatedBlock = BlockInfo.builder()
                    .fileId(block.getFileId())
                    .blkIdx(block.getBlkIdx())
                    .size(block.getSize())
                    .replicas(allReplicas)
                    .build();
            
            metadataStore.putBlock(block.getFileId(), block.getBlkIdx(), updatedBlock);
        } catch (Exception e) {
            log.error("Failed to update block metadata after removing excess replicas for {}:{}", 
                    block.getFileId(), block.getBlkIdx(), e);
        }
    }
    
    /**
     * 选择目标DataServer
     * 根据负载和排除条件选择合适的DataServer用于块复制
     *
     * @param availableDs 可用的DataServer列表
     * @param excludeDsIds 需要排除的DataServer ID集合
     * @param count 需要选择的DataServer数量
     * @param fileId 文件ID（用于日志记录）
     * @param blkIdx 块索引（用于日志记录）
     * @return 选择的DataServer列表
     */
    private List<DataServerInfo> selectTargetDataServers(List<DataServerInfo> availableDs, 
                                                        Set<String> excludeDsIds, int count, 
                                                        String fileId, int blkIdx) {
        List<DataServerInfo> candidates = availableDs.stream()
                .filter(ds -> !excludeDsIds.contains(ds.getDsId()))
                .collect(Collectors.toList());
        
        // 按负载排序（优先选择使用率较低的服务器）
        candidates.sort(Comparator.comparingDouble(DataServerInfo::getUsageRatio));
        
        return candidates.stream().limit(count).collect(Collectors.toList());
    }
    
    /**
     * 根据ID查找DataServer
     *
     * @param dsId DataServer ID
     * @param dataServers DataServer列表
     * @return 找到的DataServer，如果未找到则返回null
     */
    private DataServerInfo findDataServerById(String dsId, List<DataServerInfo> dataServers) {
        return dataServers.stream()
                .filter(ds -> ds.getDsId().equals(dsId))
                .findFirst()
                .orElse(null);
    }
    
    /**
     * 获取最新的FSCK统计信息
     * 返回最近一次执行的FSCK操作的统计结果
     *
     * @return 最新的FSCK统计，如果从未执行过FSCK则可能为null
     */
    @Override
    public IFsckService.FsckStats getLatestFsckStats() {
        return latestStats;
    }
    
    /**
     * 获取FSCK历史记录
     * 返回所有已执行的FSCK操作的统计结果列表
     *
     * @return FSCK历史列表
     */
    @Override
    public List<IFsckService.FsckStats> getFsckHistory() {
        return new ArrayList<>(fsckHistory.values());
    }
}
